package edu.sysu.demo.rabbitmq.serial;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import edu.sysu.demo.rabbitmq.SerializeException;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.util.function.Supplier;

public abstract class KryoSerializer implements ObjectSerializer {

  public static KryoSerializer threadLocal() {
    return new ThreadLocalKryoSerializer();
  }

  public static KryoSerializer simple() {
    return new SimpleKryoSerializer();
  }

  private KryoSerializer() {
  }

  @Override
  public byte[] serialize(Object o) {
    try (ByteArrayOutputStream stream = new ByteArrayOutputStream()) {
      try (Output output = new Output(stream)) {
        kryo().get().writeClassAndObject(output, o);
        output.flush();
        return stream.toByteArray();
      }
    } catch (IOException e) {
      throw new SerializeException(e);
    }
  }

  @Override
  public <T> T deserialize(byte[] byteArray, Class<T> clazz) {
    try (ByteArrayInputStream stream = new ByteArrayInputStream(byteArray)) {
      try (Input input = new Input(stream)) {
        return clazz.cast(kryo().get().readClassAndObject(input));
      }
    } catch (IOException e) {
      throw new SerializeException(e);
    }
  }


  @Override
  public Object deserialize(byte[] byteArray) {
    try (ByteArrayInputStream stream = new ByteArrayInputStream(byteArray)) {
      try (Input input = new Input(stream)) {
        return kryo().get().readClassAndObject(input);
      }
    } catch (IOException e) {
      throw new SerializeException(e);
    }
  }

  protected abstract Supplier<Kryo> kryo();

}
